/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.io;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import maxent.mr.writable.FeatureKeysWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.util.ReflectionUtils;
import stanford.tagger.FeatureKey;

/**
 *
 * @author hadoop
 */
public class FeatureReader implements HDFSReader{

    private String inputDir;
    private Set<FeatureKey> sTemplates = new HashSet<FeatureKey>();

    public FeatureReader(String inputDir) {
        this.inputDir = inputDir;
    }

    public FeatureReader() {
    }

    public String getInputDir() {
        return inputDir;
    }

    public void setInputDir(String inputDir) {
        this.inputDir = inputDir;
    }

    public Set<FeatureKey> getsTemplates() {
        return sTemplates;
    }

    public void add(Set<FeatureKey> setFK){
        Iterator<FeatureKey> fkIt = setFK.iterator();

        while(fkIt.hasNext()){
            sTemplates.add(fkIt.next());
        }
    }

    public void read() {
        Configuration conf = new Configuration();

        SequenceFile.Reader reader = null;

        try{
            FileSystem fs = FileSystem.get(conf);
            FileStatus[] inputFiles = fs.listStatus(new Path(inputDir));


            for (int i = 0; i < inputFiles.length; i++) {
                // don't process the log files
                if (!inputFiles[i].getPath().getName().equalsIgnoreCase("_logs")) {
                    reader = new SequenceFile.Reader(fs, inputFiles[i].getPath(), conf);

                    IntWritable key = (IntWritable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
                    FeatureKeysWritable value = (FeatureKeysWritable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
                    long position = reader.getPosition();

                    while(reader.next(key,value)){
                        String syncSeen = reader.syncSeen()? "*" : "";
                        System.out.printf("[%s%s]\tkey:%s\tvalue:%s\n", position, syncSeen, key, value);
                        
                        add(value.getFeatureKeys());
                        position = reader.getPosition();
                    }
                }
            }
            
        }catch(IOException ex){
            ex.printStackTrace();
        }
        finally{
            IOUtils.closeStream(reader);
        }
    }

    public void print() {

        Iterator<FeatureKey> fIt = sTemplates.iterator();
        FeatureKey fk = new FeatureKey();

        while(fIt.hasNext()){
            fk = fIt.next();
            System.out.println(fk.getNum()+":\t"+fk.getVal()+"\t"+fk.getTag());
        }
    }

}
